/**
 * 
 */
package org.tcp.client.pool.imp;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.io.UnsupportedEncodingException;
import java.util.Date;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.tcp.client.exception.PoolException;
import org.tcp.client.pool.ITcpConnection;
import org.tcp.client.pool.Pool;
import org.tcp.client.pool.PoolConfig;
import org.tcp.client.pool.PoolConfig.PoolConfigEnum;
import org.tcp.client.pool.PoolHeat;

/**
 * 实现连接的心跳检查
 * 
 * @author kucheng.liao@autonavi.com
 * @Date Oct 8, 2014
 */
public class PoolHeatImpl implements PoolHeat {

	protected static Log log = LogFactory.getLog(PoolHeatImpl.class);

	BlockingQueue<ITcpConnection> queue;

	long heat_time;
	byte[] message;
	int heatTestTime;

	public PoolHeatImpl(PoolConfig config, BlockingQueue<ITcpConnection> queue) {
		heat_time = Long.valueOf(config.getConfig().get(PoolConfigEnum.HEAT_TIME));
		heatTestTime = Integer.parseInt(config.getConfig().get(PoolConfigEnum.HEAT_TEST_TIME));
		try {
			message = config.getConfig().get(PoolConfigEnum.HEAT_PRO).getBytes("UTF-8");
		} catch (UnsupportedEncodingException e) {
			throw new PoolException(e);
		}
		this.queue = queue;
	}

	/**
	 * 心跳
	 * 
	 * @author kucheng.liao@autonavi.com
	 * @Date Oct 9, 2014
	 */
	private class PoolHeatJump extends Thread {

		public void run() {
			if (null == queue) {
				return;
			}
			while (true) {
				check();
			}
		}

		@SuppressWarnings("static-access")
		private void check() {
			try {
				ITcpConnection connection = queue.peek();
				if (null == connection) {
					return;
				}
				long currentTime = new Date().getTime();
				// 判断队列最顶端的数据最后一次的使用时间是否超过心跳时间
				if (connection.getLastUseDate().getTime() + heat_time <= currentTime) {
					// 获取队列顶端的连接，再次判断心跳时间
					connection = queue.poll(1, TimeUnit.SECONDS);

					if (log.isDebugEnabled()) {
						log.debug(" pool " + connection.pool());
					}

					if (null != connection) {
						check(connection);

						check();
					}
				}
				Thread.currentThread().sleep(heat_time);
			} catch (Exception e) {
				log.error("", e);
			}
		}
		byte[] buff = new byte[message.length];
		/**
		 * send the heat data to server
		 * 
		 * @param connection
		 */
		private void check(ITcpConnection connection) {
			int testTime = connection.getHeatTestTime();
			if (log.isDebugEnabled())
				log.debug("      heat test time " + testTime);
			if (testTime >= heatTestTime) {
				Pool pool = connection.pool();
				if (pool.getTotalSize() > pool.getMinSize()) {
					try {
						connection.destory();
						pool.addSize(-1);
					} catch (IOException e) {
						log.error("", e);
					}
					pool.decrementTotal();
					return;
				}
			}
			try {
				OutputStream out = connection.getOutputStream();
				out.write(message);
				out.flush();
				InputStream in = connection.getInputStream();
                 in.read(buff);
				if (log.isDebugEnabled()) {
					log.debug("receive heat jupm data : " + new String(buff, "UTF-8"));
				}
				connection.addHeatTestTime(1);
				connection.setLastUserDate(new Date());
				queue.add(connection);

			} catch (IOException e) {
				log.error("", e);
				Pool pool = connection.pool();
				pool.decrementTotal();
			}
		}
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see org.tcp.client.pool.PoolHeat#check()
	 */
	public void check() {

		PoolHeatJump heat = new PoolHeatJump();
		heat.start();
	}

}
